package com.ek.service.mina;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.ArrayBlockingQueue;

import org.apache.mina.core.future.CloseFuture;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.transport.socket.nio.NioDatagramAcceptor;
import org.apache.mina.transport.socket.nio.NioDatagramConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ek.common.Context;
import com.ek.common.FutureResponse;
import com.ek.common.MessageSender;
import com.ek.kademlia.protocol.Message;
import com.ek.kademlia.protocol.Response;
import com.ek.service.mina.handler.DispatcherHandler;

public class ComunicationService extends IoHandlerAdapter implements MessageSender {
    public static final int PORT = 4662;
    public static final int MULTICAST_PORT = 4682;

    private static final transient Logger log = LoggerFactory.getLogger(ComunicationService.class);
    private final Context context;
    private final DispatcherHandler handler;
    private final NioDatagramAcceptor acceptor;
    private final InetAddress group;
    private final DatagramSocket datagramSocket;
    private final NioDatagramConnector connector;

    public ComunicationService(Context context) throws Exception {
        this.context = context;
        // TODO:insert in a configuration object
        group = InetAddress.getByName("230.0.0.1");
        datagramSocket = new DatagramSocket();
        datagramSocket.setReuseAddress(false);
        handler = new DispatcherHandler();
        handler.setHandlers(context.getHandlers());
        this.context.setReceivedPacketHandler(handler);

        connector = new NioDatagramConnector();
        // if (log.isDebugEnabled()) {
        // connector.getFilterChain().addLast("connectorLogger", new LoggingFilter());
        // }
        connector.getFilterChain().addLast("connectorProtocol", new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
        connector.setHandler(this);

        acceptor = new NioDatagramAcceptor();
        // if (log.isDebugEnabled()) {
        // acceptor.getFilterChain().addLast("acceptorLogger", new LoggingFilter());
        // }
        acceptor.getFilterChain().addLast("acceptorProtocol", new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
        acceptor.getSessionConfig().setReuseAddress(true);

        acceptor.setHandler(this.handler);
        acceptor.bind(new InetSocketAddress(PORT));

    }

    public synchronized FutureResponse sendMessage(Message message, SocketAddress socketAddress) {
        log.debug("send message " + message + " to " + socketAddress);
        ArrayBlockingQueue<Response> listener = new ArrayBlockingQueue<Response>(1);
        String key = new String(message.getMessageID());
        handler.putListener(key, listener);
        send(message, socketAddress);
        return new FutureResponse(key, listener, context);
    }

    public synchronized void sendResponse(Message message, SocketAddress socketAddress) {
        log.debug("send response " + message + " to " + socketAddress);
        send(message, socketAddress);
    }

    public synchronized void sendMulticastMessage() throws IOException {
        log.debug("sending MulticastMessage");
        byte[] buf = new byte[1];
        DatagramPacket packet = new DatagramPacket(buf, buf.length, group, MULTICAST_PORT);
        datagramSocket.send(packet);
        log.debug("sended");
    }
    
    public void stop() {
        acceptor.unbind();
    }
    
    private void send(final Message message, SocketAddress socketAddress) {
        synchronized (connector) {
            ConnectFuture connected = connector.connect(socketAddress);
            connected.awaitUninterruptibly();
            connected.addListener(new IoFutureListener<ConnectFuture>() {
                public void operationComplete(ConnectFuture future) {
                    if (future.isConnected()) {
                        log.debug("connected");
                        log.debug("writing message:{}", message);
                        future.getSession().write(message);
                        log.debug("message written:{}", message);
                    } else {
                        log.error("Not connected...exiting");
                    }
                }
            });
            CloseFuture close = connected.getSession().close(false);
        }
    }
    
}
